package com.chixing.service.impl;

import com.alibaba.fastjson.JSON;
import com.alibaba.otter.canal.client.CanalConnector;
import com.alibaba.otter.canal.client.CanalConnectors;
import com.alibaba.otter.canal.protocol.CanalEntry;
import com.alibaba.otter.canal.protocol.Message;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.CommandLineRunner;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.stereotype.Component;

import java.net.InetSocketAddress;
import java.util.List;
import java.util.Map;
import java.util.HashMap;

/**
 * 通用 Canal -> Redis 同步器
 * 支持整个数据库多表同步
 */
@Component
public class CanalRedisSyncProd implements CommandLineRunner {

    private static final Logger logger = LoggerFactory.getLogger(CanalRedisSyncProd.class);

    @Autowired
    private StringRedisTemplate redisTemplate;

    private static final String CANAL_HOST = "127.0.0.1";
    private static final int CANAL_PORT = 11111;
    private static final String CANAL_DESTINATION = "example";
    private static final String CANAL_USERNAME = "canal";
    private static final String CANAL_PASSWORD = "canal_password";

    // 订阅整个数据库
    private static final String SUBSCRIBE_TABLE = "hoteldb\\..*";

    @Override
    public void run(String... args) {
        new Thread(this::startCanalSync, "Canal-Redis-Sync-Thread").start();
    }

    private void startCanalSync() {
        CanalConnector connector = null;
        try {
            connector = CanalConnectors.newSingleConnector(
                    new InetSocketAddress(CANAL_HOST, CANAL_PORT),
                    CANAL_DESTINATION,
                    CANAL_USERNAME,
                    CANAL_PASSWORD
            );

            connector.connect();
            connector.subscribe(SUBSCRIBE_TABLE);
            logger.info("✅ Canal 客户端已连接并订阅 {}", SUBSCRIBE_TABLE);

            while (true) {
                try {
                    Message message = connector.getWithoutAck(100);
                    long batchId = message.getId();
                    List<CanalEntry.Entry> entries = message.getEntries();

                    if (batchId == -1 || entries.isEmpty()) {
                        Thread.sleep(500);
                        continue;
                    }

                    for (CanalEntry.Entry entry : entries) {
                        if (entry.getEntryType() != CanalEntry.EntryType.ROWDATA) continue;
                        processEntry(entry);
                    }

                    connector.ack(batchId);

                } catch (Exception e) {
                    logger.error("❌ Canal 消费异常，rollback 当前 batch", e);
                    connector.rollback();
                    Thread.sleep(2000);
                }
            }

        } catch (Exception e) {
            logger.error("❌ Canal 主连接异常，需要重连", e);
        } finally {
            if (connector != null) {
                try { connector.disconnect(); } catch (Exception ignored) {}
            }
        }
    }

    private void processEntry(CanalEntry.Entry entry) throws Exception {
        CanalEntry.RowChange rowChange = CanalEntry.RowChange.parseFrom(entry.getStoreValue());
        String tableName = entry.getHeader().getTableName();

        for (CanalEntry.RowData rowData : rowChange.getRowDatasList()) {
            switch (rowChange.getEventType()) {
                case INSERT:
                case UPDATE:
                    Map<String, Object> data = new HashMap<>();
                    String primaryKeyValue = null;

                    for (CanalEntry.Column col : rowData.getAfterColumnsList()) {
                        data.put(col.getName(), col.getValue());
                        if (col.getIsKey()) primaryKeyValue = col.getValue();
                    }

                    if (primaryKeyValue == null) {
                        logger.warn("表 {} 找不到主键列，跳过同步", tableName);
                        continue;
                    }

                    String key = tableName + ":" + primaryKeyValue;
                    redisTemplate.opsForValue().set(key, JSON.toJSONString(data));
                    logger.info("同步到 Redis -> {}", key);
                    break;

                case DELETE:
                    String delKey = null;
                    for (CanalEntry.Column col : rowData.getBeforeColumnsList()) {
                        if (col.getIsKey()) {
                            delKey = tableName + ":" + col.getValue();
                            break;
                        }
                    }
                    if (delKey != null) {
                        redisTemplate.delete(delKey);
                        logger.info("从 Redis 删除 -> {}", delKey);
                    }
                    break;

                default:
                    break;
            }
        }
    }
}
